package yz.hdfs;

import com.mysql.jdbc.Driver;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.net.URI;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class demo8MysqlToHDFS {
    public static void main(String[] args) throws Exception{

        //创建mysql对象,

        //这是通过反射的机制来创建
        //1
        Class.forName("com.mysql.jdbc.Driver");

        //2
        //DriverManager.registerDriver(new Driver());

        Connection connection = DriverManager.getConnection("jdbc:mysql://master:3306/goods?useSSL=false&useUnicode=true&characterEncoding=utf8",
                "root","123456");

        Configuration entries = new Configuration();
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://master:9001"), entries,"root");

        //fsDataOutputStream父类中有OutputStream
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/api/jd.txt"));
        OutputStreamWriter outputStreamWriter = new OutputStreamWriter(fsDataOutputStream);

        Statement statement = connection.createStatement();
        ResultSet resultSet = statement.executeQuery("select * from jd_goods");
        while (resultSet.next()){
            String string1 = resultSet.getString(1);
            String string2 = resultSet.getString(2);
            String string3 = resultSet.getString(3);
            String string4 = resultSet.getString(4);
            String string5 = resultSet.getString(5);
            //将写出的数据写出到HDFS中
            outputStreamWriter.write(string1+","+string2+","+string3+","+string4+","+string5+"\n");
            outputStreamWriter.flush();
        }
        //关闭流操作
        statement.close();
        connection.close();
        fsDataOutputStream.close();
        outputStreamWriter.close();
        fileSystem.close();



    }
}
